package com.example.rxandroid;

import android.Manifest;
import android.content.pm.PackageManager;
import android.os.Bundle;
import android.os.Environment;
import android.os.SystemClock;
import android.util.Log;

import androidx.appcompat.app.AppCompatActivity;
import androidx.core.app.ActivityCompat;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import io.reactivex.Flowable;
import io.reactivex.FlowableSubscriber;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

public class MainActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        if (ActivityCompat.checkSelfPermission(this, Manifest.permission.READ_EXTERNAL_STORAGE) != PackageManager.PERMISSION_GRANTED) {
            ActivityCompat.requestPermissions(this, new String[]{Manifest.permission.READ_EXTERNAL_STORAGE}, 0);
        }

        ExecutorService executorService = Executors.newSingleThreadExecutor();


//        ViewModelProvider viewModelProvider = ViewModelProviders.of(this);
//        MainViewModel mainViewModel = viewModelProvider.get(MainViewModel.class);
//        mainViewModel.getMessage().subscribe(new FlowableSubscriber<String>() {
//            @Override
//            public void onSubscribe(Subscription s) {
//
//            }
//
//            @Override
//            public void onNext(String s) {
//                Log.e("androidLog","s="+s);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//
//            }
//
//            @Override
//            public void onComplete() {
//
//            }
//        });
//
//        new Thread(new Runnable() {
//            int n=1;
//            @Override
//            public void run() {
//                while (true){
//                    n++;
//                    mainViewModel.addMessage(n+"");
//                    SystemClock.sleep(3000);
//                }
//            }
//        }).start();

        test1();

    }

//


    List<String> list = new ArrayList<>();

    private final CompositeDisposable mDisposable = new CompositeDisposable();

    Observable observable;
    ObservableEmitter<String> observableEmitter;

    public Observable<String> getObservable() {
        if (observable == null) {
            observable = Observable.just(list);
        }
        return observable;
    }

    public void setObservable(Observable observable) {
        this.observable = observable;
    }


    private void test1() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("111");
                emitter.onNext("222");
                emitter.onNext("333");
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {

            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e("androidLog", "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(String s) {
                Log.e("androidLog", "onNext s=" + s);
//                if(s.equals("222")){
//                    disposable.dispose();
//                }
            }

            @Override
            public void onError(Throwable e) {
                Log.e("androidLog", "onError");
            }

            @Override
            public void onComplete() {
                Log.e("androidLog", "onComplete");
            }
        });
    }

    private void test2() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
                .subscribe(new Observer<Integer>() {

                    Disposable disposable;

                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("androidLog", "onSubscribe");
                        disposable = d;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("androidLog", integer + "");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("androidLog", "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("androidLog", "onComplete");
                    }
                });
    }


    private void test3() {

        Observable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)
                .subscribe(new Observer<Integer>() {
                    Disposable disposable;

                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("androidLog", "onSubscribe");
                        disposable = d;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("androidLog", integer + "");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("androidLog", "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("androidLog", "onComplete");
                    }
                });

    }

    private void test4() {
        Observable.fromPublisher(new Flowable<Integer>() {
            @Override
            protected void subscribeActual(Subscriber s) {
                s.onNext(1);
                s.onNext(2);
                s.onNext(3);
                s.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e("androidLog", "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.e("androidLog", integer + "");
            }

            @Override
            public void onError(Throwable e) {
                Log.e("androidLog", "onError");
            }

            @Override
            public void onComplete() {
                Log.e("androidLog", "onComplete");
            }
        });
    }

    private void test5() {

        String basePath = Environment.getExternalStorageDirectory() + File.separator;
        String files[] = {basePath + "Android", basePath + "DCIM"};
        Observable.fromIterable(Arrays.asList(files))
                .flatMap(new Function<String, ObservableSource<File>>() {
                    @Override
                    public ObservableSource<File> apply(String s) throws Exception {
                        Log.e("androidLog", s + "==============");
                        return Observable.fromIterable(Arrays.asList(new File(s).listFiles()));
                    }
                })
                .subscribe(new Observer<File>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(File file) {
                        Log.e("androidLog", file.getAbsolutePath());
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void test7() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("111");
                emitter.onNext("222");
                emitter.onNext("333");
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {

            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e("androidLog", "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(String s) {
                Log.e("androidLog", "onNext s=" + s);
//                if(s.equals("222")){
//                    disposable.dispose();
//                }
            }

            @Override
            public void onError(Throwable e) {
                Log.e("androidLog", "onError");
            }

            @Override
            public void onComplete() {
                Log.e("androidLog", "onComplete");
            }
        });
    }

    private void test8() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
                .subscribe(new Observer<Integer>() {

                    Disposable disposable;

                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("androidLog", "onSubscribe");
                        disposable = d;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("androidLog", integer + "");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("androidLog", "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("androidLog", "onComplete");
                    }
                });
    }


    private void test9() {

        Observable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)
                .subscribe(new Observer<Integer>() {
                    Disposable disposable;

                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("androidLog", "onSubscribe");
                        disposable = d;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("androidLog", integer + "");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("androidLog", "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("androidLog", "onComplete");
                    }
                });

    }

    private void test10() {
        Observable.fromPublisher(new Flowable<Integer>() {
            @Override
            protected void subscribeActual(Subscriber s) {
                s.onNext(1);
                s.onNext(2);
                s.onNext(3);
                s.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e("androidLog", "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.e("androidLog", integer + "");
            }

            @Override
            public void onError(Throwable e) {
                Log.e("androidLog", "onError");
            }

            @Override
            public void onComplete() {
                Log.e("androidLog", "onComplete");
            }
        });
    }

    private void test11() {
        Callable<String> callable = new Callable() {
            @Override
            public String call() throws Exception {
                Log.e("androidLog", "Callable:call()");
                Thread.sleep(5000);
                return "hello";
            }
        };

        Observable.fromCallable(callable)
                .subscribe(new Observer<String>() {
                    Disposable disposable;

                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("androidLog", "onSubscribe");
                        disposable = d;
                    }

                    @Override
                    public void onNext(String integer) {
                        Log.e("androidLog", integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e("androidLog", "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e("androidLog", "onComplete");
                    }
                });

    }

    private void test12() {

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("123");
                emitter.onNext("456");
                emitter.onNext("789");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e("androidLog", s);
            }
        });

    }


    private void test13() {

        class Persion {
            String name;

            public String getName() {
                return name;
            }

            public void setName(String name) {
                this.name = name;
            }

            public Persion(String name) {
                this.name = name;
            }
        }

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("张三");
                emitter.onNext("李四");
                emitter.onNext("王五");
            }
        }).map(new Function<String, Persion>() {
            @Override
            public Persion apply(String s) throws Exception {
                return new Persion(s);
            }
        }).subscribe(new Observer<Persion>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Persion persion) {
                Log.e("androidLog", persion.getName());
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void test14() {

        Observable observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("111");
                emitter.onNext("222");
                emitter.onNext("333");
                emitter.onNext("777");
            }
        });

        Observable observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("444");
                emitter.onNext("555");
                emitter.onNext("666");

            }
        });

        Observable.zip(observable1, observable2, new BiFunction<String, String, String>() {
            @Override
            public String apply(String o, String o2) throws Exception {
                return o + "==" + o2;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String o) {
                Log.e("androidLog", o);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void test15() {

        Observable observable1 = Observable.just(1, 2, 3);

        Observable observable2 = Observable.just(4, 5, 6);

        Observable observable3 = Observable.just(7, 8, 9);

        Observable.concat(observable1, observable2, observable3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer o) {
                        Log.e("androidLog", o + "");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });


    }


    private void test16() {

        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
                .buffer(3)
                .subscribe(new Observer<List<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        for (int n : integers) {
                            Log.e("androidLog", n + "");
                        }
                        Log.e("androidLog", "=========================");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void test17() {

        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer != 2;
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("androidLog", integer + "");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
}
